package com.tang.window;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 全量窗口
 *
 * @author tang
 * @since 2023/6/11 13:15
 */
public class WindowProcessDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> source = env
                .socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedStream = source.keyBy(WaterSensor::getId);

        // 1。窗口分配器
        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        // 老写法
//        window.apply(new WindowFunction<WaterSensor, String, String, TimeWindow>() {
//            /**
//             * 全量窗口函数
//             * 
//             * @param s The key for which this window is evaluated. 分组的Key
//             * @param window The window that is being evaluated. 窗口对象
//             * @param input The elements in the window being evaluated. 存入的数据
//             * @param out A collector for emitting elements. 采集器
//             * @throws Exception -
//             */
//            @Override
//            public void apply(String s, TimeWindow window, Iterable<WaterSensor> input, Collector<String> out) throws Exception {
//                
//            }
//        }).print();

        SingleOutputStreamOperator<String> process = window.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
            /**
             * 全窗口函数计算逻辑：  窗口触发时才会调用一次，统一计算窗口的所有数据
             * @param s   分组的key
             * @param context  上下文
             * @param elements 存的数据
             * @param out      采集器
             * @throws Exception -
             */
            @Override
            public void process(String s, ProcessWindowFunction<WaterSensor, String, String
                    , TimeWindow>.Context context, Iterable<WaterSensor> elements
                    , Collector<String> out) throws Exception {
                long start = context.window().getStart();
                long end = context.window().getEnd();
                String pattern = "yyyy-MM-dd HH:mm:ss.SSS";
                String windowStart = DateFormatUtils.format(start, pattern);
                String windowEnd = DateFormatUtils.format(end, pattern);

                long count = elements.spliterator().estimateSize();
                out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含"
                        + count + "条数据====>" + elements);
            }
        });

        process.print();

        env.execute();
    }

}
